<?php

namespace App\Console\Commands\Nsq;

use App\Library\FeiShu\SendFs;
use App\Models\Common\ConfigModel;
use Illuminate\Console\Command;
use Illuminate\Support\Facades\Cache;
use Illuminate\Support\Facades\Http;
use Illuminate\Support\Facades\Log;
use Kuafu\GeneralFs\BuildFs;

class MonitorNsqServerCommand extends Command
{
    protected $signature = 'nsq:monitor-nsq-server';

    protected $description = '监控nsq服务信息';

    private array $lastTopics = [];
    private int   $lastTimestamps;

    private array $errs = [];

    public function handle()
    {
        $str = $this->do();
        if ($str) {
            BuildFs::newRobot(SendFs::ROBOT_NSQ_MONITOR)
                   ->setTitleContent('NSQ监控预警 - ' . config('app.env'), $str)
                   ->send();

            $this->info('NSQ监控预警 - ' . config('app.env'));
            $this->info($str);
        }
    }

    private function do()
    {
        $url = str_replace('4161', '4151', config("nsq.default.address"));

        // ping
        $ping = (array)Http::withOptions(['http_errors' => false])
                           ->get(rtrim($url, '/') . '/ping?format=json')
                           ->body();
        if (!isset($ping[0]) || $ping[0] != 'OK') {
            Log::error('nsq服务ping失败');
            return 'nsq服务ping失败';
        }

        // 监控信息
        $stats = (array)Http::withOptions(['http_errors' => false])
                            ->get(rtrim($url, '/') . '/stats?format=json')
                            ->json();

        $health = $stats['health'] ?? '';
        $topics = $stats['topics'] ?? [];

        if (empty($stats) || $health != 'OK' || !$topics) {
            Log::error('nsq服务异常，未正确获取到stats');
            return 'nsq服务异常，未正确获取到stats';
        }

        // 上次的topics，keyBy格式化
        $this->lastTopics();

        // 存储新的topics
        $this->setTopics($topics);

        if (!$this->lastTopics) {
            return ''; // 首次检测不做处理
        }

        foreach ($topics as $topic) {
            $channels   = $topic['channels'] ?? [];
            $topic_name = $topic['topic_name'] ?? '';

            // 没有订阅的客户端，先不管
//            if (!$channels) {
//            }

            // 检查channel
            foreach ($channels as $channel) {
                $this->checkChannel($topic_name, $channel);
            }
        }

        // 处理errs
        return $this->errsFs();
    }

    /**
     * 检查channel
     * @param array $channel
     * @return void
     */
    private function checkChannel(string $topic_name, array $channel): void
    {
        // 客户端数量，=0的先不管
        $client_count = $channel['client_count'] ?? 0;
        if ($client_count <= 0) {
            return;
        }

        // 消费端名称
        $channel_name = $channel['channel_name'] ?? '';

        // 待消费的消息数量
        $depth = $channel['depth'] ?? 0;

        // 已消费的消息数量
        $message_count = $channel['message_count'] ?? 0;

        // 消费者数量
        $client_count = $channel['client_count'] ?? 0;

        // 堵塞的阈值
        $threshold = $this->threshold($topic_name);
        if ($depth >= $threshold) {

            // 上次存档的信息
            $last_depth         = $this->lastTopics[$topic_name]['channels'][$channel_name]['depth'] ?? 0;
            $last_message_count = $this->lastTopics[$topic_name]['channels'][$channel_name]['message_count'] ?? 0;

            // 期间新增的消费数量
            $diff_message_count = $message_count - $last_message_count;
            $diff_seconds       = time() - $this->lastTimestamps;
            $diff_depth         = $depth - $last_depth;

            $this->errs[] = [
                'topic_name'         => $topic_name,
                'channel_name'       => $channel_name,
                'threshold'          => $threshold,
                'depth'              => $depth,
                'client_count'       => $client_count,
                'diff_seconds'       => $diff_seconds,
                'diff_message_count' => $diff_message_count,
                'diff_depth'         => $diff_depth,
            ];
        }
    }

    /**
     * 记录当前的topics
     * @param array $topics
     * @return bool
     */
    private function setTopics(array $topics)
    {
        return Cache::set('MonitorNsqServerCommand:topics', [time(), $topics], 86400);
    }

    /**
     * 上次的topics，格式化
     * @return array
     */
    private function lastTopics(): void
    {
        list($this->lastTimestamps, $lastTopicsTmp) = Cache::get('MonitorNsqServerCommand:topics') ?? [0, []];

        foreach ($lastTopicsTmp as $topicTmp) {
            $this->lastTopics[$topicTmp['topic_name']] = $topicTmp;

            unset($this->lastTopics[$topicTmp['topic_name']]['channels']);

            foreach ($topicTmp['channels'] as $channel) {
                $this->lastTopics[$topicTmp['topic_name']]['channels'][$channel['channel_name']] = $channel;
            }
        }
    }

    /**
     * 阈值
     * @param string $topic_name
     * @return int
     */
    private function threshold(string $topic_name): int
    {
        $configs = Cache::store('array')->remember('MonitorNsqServerCommand:threshold', 3600, function () {
            return ConfigModel::getValueNewJson('nsq', 'monitor-nsq-server-threshold');
        });

        if (isset($configs[$topic_name])) {
            return $configs[$topic_name];
        }

        if (isset($configs['default'])) {
            return $configs['default'];
        }

        return 100;
    }

    /**
     * 异常通知
     * @return string
     */
    private function errsFs(): string
    {
        if (!$this->errs) {
            return '';
        }

        $str = '';
        foreach ($this->errs as $err) {
            $str .= "topic_name: {$err['topic_name']}" . PHP_EOL;
            $str .= "channel_name: {$err['channel_name']}" . PHP_EOL;
            $str .= "当前积压数量: {$err['depth']}" . PHP_EOL;
            $str .= "队列积压阈值: {$err['threshold']}" . PHP_EOL;
            $str .= "消费进程数: {$err['client_count']}" . PHP_EOL;
            $str .= "距离上次检测时间: {$err['diff_seconds']}秒" . PHP_EOL;
            $str .= "距离上次检测消费的数量: {$err['diff_message_count']}" . PHP_EOL;
            $str .= "距离上次积压数量的变化: {$err['diff_depth']}" . PHP_EOL;
            $str .= PHP_EOL;
            $str .= PHP_EOL;
        }

        return $str;
    }
}
